package com.demo.function.thread;

import org.apache.commons.lang3.concurrent.BasicThreadFactory;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 测试线程池不同的BlockingQueue
 * @author K
 */
public class ThreadQueue {

    /**
     * 测试通过newFixedThreadPool创建线程池，BlockingQueue使用LinkedBlockingQueue
     * 使用LinkedBlockingQueue，线程池数量等于corePoolSize，maximumPoolSize。
     * 当并发执行任务大于线程池数量，待处理任务会加入LinkedBlockingQueue，等待空闲线程处理。
     * LinkedBlockingQueue是无界任务队列，它是基于链表的先进先出队列，如果创建是没有指定队列大小，则默认值为Integer.MAX_VALUE
     */
    public void linkedBlockingQueue() {
        BasicThreadFactory threadFactory = new BasicThreadFactory.Builder()
                .namingPattern("basicThreadFactory-").build();

        ExecutorService executorService = new ThreadPoolExecutor(10, 10,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(), threadFactory);

        int length = 500;
        CountDownLatch countDownLatch = new CountDownLatch(1);
        for (int i = 0; i < length; i++) {
            executorService.execute(new TestRunable(i, countDownLatch));
        }
        countDownLatch.countDown();
    }

    /**
     * 测试使用ArrayBlockingQueue创建线程池
     * 当并发执行任务大于线程池数量，待处理任务会加入ArrayBlockingQueue。
     * 如果待处理任务数量大于ArrayBlockingQueue长度，则发生溢出异常。
     * ArrayBlockingQueue是有界队列，基于数组的先进先出队列，创建时必须指定大小
     */
    public void arrayBlockingQueue(int threadNums) {
        ExecutorService executorService = new ThreadPoolExecutor(10, 20,
                0L, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(10));

        CountDownLatch countDownLatch = new CountDownLatch(1);
        for (int i = 0; i < threadNums; i++) {
            executorService.execute(new TestRunable(i, countDownLatch));
        }
        countDownLatch.countDown();
    }

    /**
     * 测试使用SynchronousQueue创建线程池
     * SynchronousQueue是直接提交队列，它不会保存提交任务，而是直接新建一条线程来执行新任务
     */
    public void synchronousQueue() {
        ExecutorService executorService = new ThreadPoolExecutor(0, 11,
                0L, TimeUnit.MILLISECONDS,
                new SynchronousQueue<>());

        int length = 11;
        CountDownLatch countDownLatch = new CountDownLatch(1);
        for (int i = 0; i < length; i++) {
            executorService.execute(new TestRunable(i, countDownLatch));
        }
        countDownLatch.countDown();
    }

    /**
     * 测试通过newScheduledThreadPool创建线程池，BlockingQueue使用DelayedWorkQueue
     * DelayedWorkQueue配合ScheduledThreadPoolExecutor，实现定时执行任务
     */
    public void delayQueue() {
        ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1);

        int length = 5;
        for (int i = 0; i < length; i++) {
            executorService.scheduleAtFixedRate(new ScheduleRunnable(i), 1,1, TimeUnit.SECONDS);
        }
    }
}
